// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <cantor/catalog/database_manager.h>
#include <cantor/catalog/namespace_manager.h>
#include <cantor/catalog/namespace_querier.h>

namespace cantor::catalog {

    turbo::Status DatabaseManager::create_database(std::shared_ptr<DatabaseInfo> info, CatalogStorage *storage,
                                      NamespaceManager *ns_manager) {
        int64_t ns_id = -1;
        int64_t temp_id = -1;
        {
            std::unique_lock lk(_mutex);

            auto it = _database_id_map.find(info->database_name());
            if (it != _database_id_map.end()) {
                return turbo::already_exists_error("database ", info->database_name(), "already exists with id:", it->second);
            }
            NamespaceQuerier ns_q(ns_manager);

            TURBO_MOVE_OR_RAISE(ns_id, ns_q.namespace_id(info->namespace_name()));

            info->set_namespace_id(ns_id);
            /// save to storage
            temp_id = _max_database_id + 1;
            info->set_database_id(temp_id);
            info->upgrade_version();
            auto key = storage->make_database_key(info->database_name(), temp_id);
           TURBO_MOVE_OR_RAISE( auto value , info->serialize());
            TURBO_RETURN_NOT_OK(storage->save_database(key, value, temp_id));
            /// update memory
            _max_database_id = temp_id;
            _database_id_map[info->database_name()] = temp_id;
            _database_map[temp_id] = std::move(info);
        }
        {
            return ns_manager->add_database_id(ns_id, temp_id);
        }
    }

    turbo::Status DatabaseManager::remove_database(const std::string &db_name, CatalogStorage *storage,
                                      NamespaceManager *ns_manager) {
        int64_t ns_id = -1;
        int64_t temp_id = -1;
        {
            std::unique_lock lk(_mutex);

            auto it = _database_id_map.find(db_name);
            if (it == _database_id_map.end()) {
                return turbo::not_found_error("database ", db_name, " not found");
            }

            temp_id = it->second;

            auto tables_it = _table_ids.find(it->second);
            if(!tables_it->second.empty()) {
                return turbo::failed_precondition_error("database ", db_name, " have tables yet, clear all tables before remove database");
            }
            auto &info = _database_map[it->second];

            NamespaceQuerier ns_q(ns_manager);

            TURBO_MOVE_OR_RAISE(ns_id, ns_q.namespace_id(info->namespace_name()));

            /// save to storage
            auto key = storage->make_database_key(info->database_name(), temp_id);
            TURBO_RETURN_NOT_OK(storage->remove_database(key));
            /// update memory
            _database_map.erase(temp_id);
            _database_id_map.erase(db_name);
            _table_ids.erase(temp_id);

        }
        {
            return ns_manager->remove_database_id(ns_id, temp_id);
        }

    }

    turbo::Status DatabaseManager::update_database(std::shared_ptr<DatabaseInfo> info, CatalogStorage *storage) {
        std::unique_lock lk(_mutex);

        auto it = _database_id_map.find(info->database_name());
        if (it == _database_id_map.end()) {
            return turbo::not_found_error("not found database ", info->database_name());
        }

        auto id = it->second;
        auto &old = _database_map[id];
        TURBO_ABORT_NOT_OK(info->merge_for_update(old.get()));
        info->upgrade_version();
        auto key = storage->make_database_key(info->database_name(), id);
        TURBO_MOVE_OR_RAISE(auto value , info->serialize());
        TURBO_RETURN_NOT_OK(storage->update_database(key, value));

        /// update memory
        _database_map[id] = std::move(info);
        return turbo::OkStatus();
    }

    void DatabaseManager::set_max_id(const int64_t max_id) {
        std::unique_lock lk(_mutex);
        _max_database_id = max_id;
    }

    turbo::Status DatabaseManager::add_table_id(int64_t db_id, int64_t table_id) {
        std::unique_lock lk(_mutex);
        auto it = _database_map.find(db_id);
        if (it == _database_map.end()) {
            return turbo::not_found_error("not found database ", db_id);
        }
        _table_ids[db_id].insert(table_id);
        return turbo::OkStatus();
    }

    turbo::Status DatabaseManager::remove_table_id(int64_t db_id, int64_t table_id) {
        std::unique_lock lk(_mutex);
        auto it = _table_ids.find(db_id);
        if (it == _table_ids.end()) {
            return turbo::not_found_error("not found database ", db_id);
        }
        _table_ids[db_id].erase(table_id);
        return turbo::OkStatus();
    }

    void DatabaseManager::clear_tables(int64_t db_id) {
        std::unique_lock lk(_mutex);
        _table_ids.erase(db_id);
    }
}  // namespace cantor::catalog
